use crate::cmp::Cmp;
use crate::key_types::{parse_internal_key, truncate_to_userkey, LookupKey, ValueType};
use crate::merging_iter::MergingIter;
use crate::snapshot::Snapshot;
use crate::types::{Direction, LdbIterator, Shared};
use crate::version_set::VersionSet;

use bytes::Bytes;
use std::cmp::Ordering;
use std::mem;
use std::rc::Rc;

/// READ_BYTES_PERIOD 是一个常量，值为 1048576（即 1MB，因为 1024×1024 = 1048576），从命名和值来看，它主要用于量化控制与 “读取字节数” 相关的周期性逻辑。
///
/// 结合你之前提到的存储引擎（如 LSM 树）场景，它的具体用途可能包括：
///
/// 作为读取操作的统计周期阈值：例如每累计读取 1MB 数据时，触发一次性能指标记录（如吞吐量计算、IO 次数统计）。
///
/// 控制后台任务的触发频率：比如在读取操作达到 1MB 时，判断是否需要执行 SSTable 压缩、缓存刷新等优化操作。
///
/// 平衡系统负载：配合随机函数（如你之前提到的 random_period）生成 0~2MB 范围的随机值，用于分散周期性任务的执行时机，避免多个任务同时触发导致的资源竞争。
///
/// 简单来说，它是一个与 “读取字节量” 相关的基准单位，用于标准化和控制存储引擎中与读取操作相关的周期性行为，确保系统高效、均衡地运行。
const READ_BYTES_PERIOD: isize = 1048576;

/// DBIterator is an iterator over the contents of a database.
///
/// 这个 DBIterator 结构体是数据库对外提供的键值对迭代器，用于遍历数据库中的键值对。它封装了 LSM 树存储引擎的复杂底层结构，
/// 为用户提供了简单、一致的遍历接口，同时确保遍历操作基于特定的快照状态（避免受并发写入影响）。
pub struct DBIterator {
    // A user comparator.
    /// 键比较器，用于定义键的排序规则（如字典序），确保迭代器按数据库指定的顺序返回键值对。
    cmp: Rc<Box<dyn Cmp>>,
    /// 指向版本集合的共享指针，包含数据库当前的所有 SSTable 元数据（如各层级文件列表），迭代器需要通过它定位数据存储位置。
    vset: Shared<VersionSet>,
    /// 合并迭代器，内部会聚合多个底层数据源（如内存表、各层级 SSTable 的迭代器），并按键的顺序合并结果，让上层只需面对一个统一的迭代流。
    /// MergingIter 负责全局有序合并，BlockIter 负责数据块内的高效遍历。
    iter: MergingIter,
    // By holding onto a snapshot, we make sure that the iterator iterates over the state at the
    // point of its creation.
    /// 迭代器关联的快照，确保遍历范围被限制在该快照对应的时间点（序列号），忽略之后的所有数据修改，保证遍历结果的一致性。
    ss: Snapshot,
    /// 迭代方向（如向前 Forward 或向后 Reverse），控制迭代器的移动方向（从第一个到最后一个，或反之）。
    dir: Direction,
    /// 记录迭代过程中已读取的字节数，用于流量控制或统计（如限制单次遍历的总数据量）。
    byte_count: isize,

    /// 标记迭代器当前是否处于有效状态（即是否指向一个有效的键值对），false 表示已遍历完所有数据。
    valid: bool,
    // temporarily stored user key.
    /// 用于缓存键的字节数据：keybuf 临时存储从底层读取的内部键（包含序列号和类型），savedkey 缓存提取出的用户键（供外部访问）。
    savedkey: Vec<u8>,
    // buffer for reading internal keys
    keybuf: Vec<u8>,
    /// 用于缓存值的字节数据：valbuf 临时存储从底层读取的值，savedval 缓存处理后的用户值（供外部访问）。
    savedval: Vec<u8>,
    valbuf: Vec<u8>,
}

impl DBIterator {
    pub fn new(
        cmp: Rc<Box<dyn Cmp>>,
        vset: Shared<VersionSet>,
        iter: MergingIter,
        ss: Snapshot,
    ) -> DBIterator {
        DBIterator {
            cmp,
            vset,
            iter,
            ss,
            dir: Direction::Forward,
            byte_count: random_period(),

            valid: false,
            savedkey: vec![],
            keybuf: vec![],
            savedval: vec![],
            valbuf: vec![],
        }
    }

    /// record_read_sample records a read sample using the current contents of self.keybuf, which
    /// should be an InternalKey.
    ///
    /// 这个 record_read_sample 方法是 LSM 树存储引擎中用于采集数据读取样本的统计工具，通过间歇性地记录读取操作的关键信息（如读取的内部键），为后续的存储优化（尤其是 SSTable 的压缩策略）提供数据支撑。
    fn record_read_sample(&mut self, len: usize) {
        // 扣减当前读取的字节数
        // self.byte_count 是迭代器维护的 “采样触发计数器”，初始值通常由随机周期初始化；
        // 每次读取一个键值对后，用总字节数（键 + 值）扣减计数器，逐步向 “触发采样” 的阈值（< 0）靠近。
        self.byte_count -= len as isize;
        // 当计数器 byte_count 小于 0 时，触发一次采样：
        // 获取当前版本：self.vset.borrow().current() 拿到数据库当前的版本（Version），版本中维护了读取样本的统计信息；
        // 记录样本：调用版本的 record_read_sample(&self.keybuf)，将当前读取的内部键（keybuf，含序列号和操作类型）存入统计；
        // 重置计数器：通过 random_period() 生成随机值，累加回 byte_count 直到其大于等于 0，确保下一次采样的触发时机是随机的，避免固定频率导致的统计偏差。
        if self.byte_count < 0 {
            // 触发采样：记录当前内部键到版本的统计中
            let v = self.vset.borrow().current();
            v.borrow_mut().record_read_sample(&self.keybuf);
            // 重置计数器：通过随机周期避免采样过于规律
            while self.byte_count < 0 {
                // 生成一个在特定范围内的随机整
                self.byte_count += random_period();
            }
        }
    }

    /// find_next_user_entry skips to the next user entry after the one saved in self.savedkey.
    ///
    /// 这个 find_next_user_entry 方法是 DBIterator 内部用于过滤无效数据并定位到下一个有效用户键值对的核心逻辑。
    /// 它会跳过被删除的键、被快照排除的新版本键以及被覆盖的旧版本键，确保迭代器最终指向用户可见的有效数据。
    ///
    /// find_next_user_entry 的核心价值是实现 LSM 树迭代中的数据过滤与有效性校验，解决了以下关键问题：
    ///
    /// 快照一致性过滤：只保留快照创建前（序列号 <= 快照序列号）的操作，确保遍历结果与快照状态一致；
    ///
    /// 删除标记处理：跳过被标记删除的键及其旧版本，避免用户看到已删除的数据；
    ///
    /// 版本覆盖处理：通过键比较跳过被新版本覆盖的旧版本键，只保留最新有效版本；
    ///
    /// 迭代状态维护：更新迭代器的有效性（valid）和缓存（savedkey、savedval），为外部提供准确的迭代状态。
    fn find_next_user_entry(&mut self, mut skipping: bool) -> bool {
        assert!(self.iter.valid());
        assert!(self.dir == Direction::Forward);
        // 从底层合并迭代器获取当前键值对（内部键 + 值）
        // 通过循环遍历底层合并迭代器（self.iter）的所有条目，逐个处理键值对；
        // 先缓存内部键和值，并解析出关键信息：
        // typ：操作类型（TypeValue 表示插入，TypeDeletion 表示删除标记）；
        // seq：该键值对的序列号（记录操作发生的时间顺序）；
        // ukey：用户原始键（已剥离内部元数据）。
        while self.iter.valid() {
            if let Some((key_bytes, val_bytes)) = self.iter.current() {
                self.keybuf = key_bytes.to_vec(); // 缓存内部键（含元数据）
                self.savedval = val_bytes.to_vec();  // 缓存值
                // 记录读取的字节数（用于统计或采样）
                let len = self.keybuf.len() + self.savedval.len();
                self.record_read_sample(len);
                // 解析内部键：提取操作类型、序列号、用户键
                let (typ, seq, ukey) = parse_internal_key(&self.keybuf);
                // ... 过滤逻辑 ...
                // Skip keys with a sequence number after our snapshot.
                // 只保留序列号 <= 快照序列号的操作（排除快照创建后的新操作）
                // 快照（self.ss）记录了迭代器应遵循的时间点（序列号），任何序列号大于快照序列号的操作都是在快照创建后发生的，应被忽略，确保遍历结果与快照状态一致。
                if seq <= self.ss.sequence() {
                    // 若操作类型是删除标记，表示该键在该序列号时被删除；
                    // 缓存被删除的键，并设置 skipping = true，后续遇到该键的旧版本时也应跳过（因为已被删除）。
                    if typ == ValueType::TypeDeletion {
                        // Mark current (deleted) key to be skipped.
                        // 缓存被删除的用户键，后续跳过该键的所有旧版本
                        self.savedkey.clear();
                        self.savedkey.extend_from_slice(ukey);
                        skipping = true; // 标记需要跳过该键
                    } else if typ == ValueType::TypeValue {
                        // 若操作类型是正常值（插入或更新），需要判断是否应被保留：
                        // 若 skipping 为 true 且当前键小于等于被删除的键（savedkey），说明该键已被后续操作删除，应忽略；
                        // 否则，该键是有效的（未被删除且在快照范围内），标记迭代器为有效状态并返回 true。
                        if skipping && self.cmp.cmp(ukey, &self.savedkey) <= Ordering::Equal {
                            // Entry hidden, because it's smaller than the key to be skipped.
                            // 若需要跳过且当前键 <= 被删除的键，说明该键已被删除或被覆盖，忽略
                        } else {
                            // 找到有效键值对：标记迭代器有效，清空缓存的待跳过键，返回成功
                            self.valid = true;
                            self.savedkey.clear();
                            return true;
                        }
                    }
                }
            }
            self.iter.advance();  // 推进到底层迭代器的下一个条目
        }
        // 循环结束（底层迭代器已无数据）
        self.savedkey.clear();
        self.valid = false;
        // 若遍历完所有底层数据仍未找到有效条目，标记迭代器为无效状态，返回 false。
        false
    }

    /// find_prev_user_entry, on a backwards-moving iterator, stores the newest non-deleted version
    /// of the entry with the key == self.savedkey that is in the current snapshot, into
    /// savedkey/savedval.
    fn find_prev_user_entry(&mut self) -> bool {
        assert!(self.dir == Direction::Reverse);
        let mut value_type = ValueType::TypeDeletion;

        // The iterator should be already set to the previous entry if this is a direction change
        // (i.e. first prev() call after advance()). savedkey is set to the key of that entry.
        //
        // We read the current entry, ignore it for comparison (because the initial value_type is
        // Deletion), assign it to savedkey and savedval and go back another step (at the end of
        // the loop).
        //
        // We repeat this until we hit the first entry with a different user key (possibly going
        // through newer versions of the same key, because the newest entry is first in order),
        // then break. The key and value of the latest entry for the desired key have been stored
        // in the previous iteration to savedkey and savedval.
        while self.iter.valid() {
            if let Some((key_bytes, val_bytes)) = self.iter.current() {
                self.keybuf = key_bytes.to_vec();
                self.valbuf = val_bytes.to_vec();
                let len = self.keybuf.len() + self.valbuf.len();
                self.record_read_sample(len);
                let (typ, seq, ukey) = parse_internal_key(&self.keybuf);

                if seq > 0 && seq <= self.ss.sequence() {
                    if value_type != ValueType::TypeDeletion
                        && self.cmp.cmp(ukey, &self.savedkey) == Ordering::Less
                    {
                        // We found a non-deleted entry for a previous key (in the previous iteration)
                        break;
                    }
                    value_type = typ;
                    if value_type == ValueType::TypeDeletion {
                        self.savedkey.clear();
                        self.savedval.clear();
                    } else {
                        self.savedkey.clear();
                        self.savedkey.extend_from_slice(ukey);

                        mem::swap(&mut self.savedval, &mut self.valbuf);
                    }
                }
            }
            self.iter.prev();
        }

        if value_type == ValueType::TypeDeletion {
            self.valid = false;
            self.savedkey.clear();
            self.savedval.clear();
            self.dir = Direction::Forward;
        } else {
            self.valid = true;
        }
        true
    }
}

impl LdbIterator for DBIterator {

    /// 这个 advance 方法是 DBIterator 内部用于推进迭代器到下一个有效键值对的核心逻辑，负责处理迭代方向切换、当前状态维护和有效数据过滤，确保迭代器始终指向向符合要求的用户可见键值对。
    ///
    /// advance 方法的核心价值是实现迭代器的安全推进，解决了 LSM 树迭代中的几个关键问题：
    ///
    /// 方向切换处理：平滑处理从反向到正向的迭代方向切换，确保状态一致；
    ///
    /// 内部键转用户键：剥离内部元数据（序列号、操作类型），向用户暴露干净的原始键；
    ///
    /// 无效数据过滤：通过 find_next_user_entry 跳过删除标记或被覆盖的旧版本，只保留有效的用户数据；
    ///
    /// 状态维护：更新缓存（savedkey、savedval）和有效性标记（valid），为外部提供一致的迭代状态查询。
    fn advance(&mut self) -> bool {
        println!("----------db advance----------");
        // 当迭代器处于无效状态（valid = false，如刚初始化或已遍历完所有数据）时，通过 seek_to_first() 重置到第一个键值对，再返回其有效性。
        if !self.valid() {
            self.seek_to_first(); // 若当前无效，重置到第一个条目
            return self.valid();  // 返回重置后的有效性
        }
        // 当迭代器之前处于反向（Reverse）模式时，先切换到正向（Forward）模式；
        // 根据底层合并迭代器（self.iter）的状态，要么重置到开头，要么推进到下一个条目；
        // 若底层迭代器已无效（无更多数据），则标记当前迭代器为无效并返回 false。
        if self.dir == Direction::Reverse {
            self.dir = Direction::Forward; // 切换方向为正向
            if !self.iter.valid() {
                self.iter.seek_to_first();  // 若底层迭代器无效，重置到开头
            } else {
                self.iter.advance(); // 否则推进到底层迭代器的下一个条目
            }
            // 检查底层迭代器是否有效
            if !self.iter.valid() {
                self.valid = false;  // 标记为无效
                self.savedkey.clear(); // 清空缓存的键
                return false;
            }
        } else {
            // 对于正向迭代，先从底层迭代器获取当前键值对；
            // 缓存键值对数据，并通过 truncate_to_userkey 处理键：LSM 树的内部键包含序列号和操作类型（如删除标记），这里需要剥离这些元数据，只保留用户传入的原始键。
            // Save current user key.
            // 保存当前用户键
            if let Some((key_bytes, val_bytes)) = self.iter.current() {
                self.savedkey = key_bytes.to_vec();  // 缓存底层键（包含内部元数据）
                self.savedval = val_bytes.to_vec();  // 缓存底层值
                truncate_to_userkey(&mut self.savedkey); // 提取纯用户键（移除内部序列号等元数据）
            } else {
                panic!("Iterator should be valid here");  // 理论上不会触发，确保状态一致性
            }
        }
        // 调用 find_next_user_entry 过滤无效条目（如被删除的键、被新版本覆盖的旧版本键）；
        // 参数 true 表示需要跳过当前键的重复版本（只保留最新版本），确保用户看到的是最终有效数据。
        self.find_next_user_entry(
            // skipping=
            true,
        )
    }
    /// 这个 current 方法是 DBIterator 结构体用于获取当前指向的键值对的核心接口。
    /// 它根据迭代方向（正向或反向）返回对应的用户可见键值对，同时处理内部键到用户键的转换，确保用户获取到的是干净、有效的数据。
    fn current(&self) -> Option<(Bytes, Bytes)> {
        println!("----------db current----------");
        // 调用 self.valid() 检查迭代器是否处于有效状态（是否指向一个可访问的键值对）；
        // 若无效（如已遍历完毕或未初始化），返回 None，避免错误访问。
        if !self.valid() {
            return None;  // 若迭代器无效，直接返回 None
        }
        // If direction is forward, savedkey and savedval are not used.
        // 正向迭代时，直接从底层合并迭代器（self.iter）获取当前键值对：
        // key_bytes 是数据库内部存储的键（含序列号和操作类型等元数据）；
        // 通过 truncate_to_userkey 处理 key_bytes，剥离元数据，提取用户原始键；
        // 将用户键和原始值包装为 Bytes 类型返回（Bytes 提供高效的内存管理，适合频繁传递）。
        if self.dir == Direction::Forward {
            if let Some((key_bytes, val_bytes)) = self.iter.current() {
                // 从底层迭代器获取内部键（含元数据），转换为用户键
                let mut key = key_bytes.to_vec();
                truncate_to_userkey(&mut key);  // 剥离8字节元数据，保留原始用户键
                // 包装为 Bytes 类型返回（Bytes 是高效的字节切片类型）
                Some((Bytes::from(key), val_bytes))
            } else {
                None
            }
        } else {
            Some((
                Bytes::copy_from_slice(&self.savedkey),  // 返回缓存的用户键
                Bytes::copy_from_slice(&self.savedval),  // 返回缓存的用户值
            ))
        }
    }
    fn prev(&mut self) -> bool {
        println!("----------db prev----------");
        if !self.valid() {
            return false;
        }

        if self.dir == Direction::Forward {
            // scan backwards until we hit a different key; then use the normal scanning procedure:
            // find_prev_user_entry() wants savedkey to be the key of the entry that is supposed to
            // be left in savedkey/savedval, which is why we have to go to the previous entry before
            // calling it.
            if let Some((key_bytes, val_bytes)) = self.iter.current() {
                self.savedkey = key_bytes.to_vec();
                self.savedval = val_bytes.to_vec();
                truncate_to_userkey(&mut self.savedkey);
            }
            loop {
                self.iter.prev();
                if !self.iter.valid() {
                    self.valid = false;
                    self.savedkey.clear();
                    self.savedval.clear();
                    return false;
                }
                // Scan until we hit the next-smaller key.
                if let Some((key_bytes, _val_bytes)) = self.iter.current() {
                    self.keybuf = key_bytes.to_vec();
                    truncate_to_userkey(&mut self.keybuf);
                }
                if self.cmp.cmp(&self.keybuf, &self.savedkey) == Ordering::Less {
                    break;
                }
            }
            self.dir = Direction::Reverse;
        }
        self.find_prev_user_entry()
    }
    fn valid(&self) -> bool {
        println!("----------db valid----------");
        self.valid
    }
    fn seek(&mut self, to: &[u8]) {
        println!("----------db seek----------");
        self.dir = Direction::Forward;
        self.savedkey.clear();
        self.savedval.clear();
        self.savedkey
            .extend_from_slice(LookupKey::new(to, self.ss.sequence()).internal_key());
        self.iter.seek(&self.savedkey);
        if self.iter.valid() {
            self.find_next_user_entry(
                // skipping=
                false,
            );
        } else {
            self.valid = false;
        }
    }

    /// 这段段代码定义了一个名为 seek_to_first 的方法，用于将迭代器定位到第一个有效用户条目的位置，是迭代器操作中的常见初始化方法。
    fn seek_to_first(&mut self) {
        println!("----------db seek_to_first----------");
        // 设置迭代方向：self.dir = Direction::Forward 将迭代方向设置为正向（从前往后遍历）。
        self.dir = Direction::Forward;
        // 清空缓存：self.savedval.clear() 清除可能缓存的上一个值，确保状态干净。
        self.savedval.clear();
        // self.iter 是 MergingIter
        // 定位到底层迭代器起点：self.iter.seek_to_first() 调用底层合并迭代器（MergingIter）的方法，
        // 将其定位到最开始的位置。MergingIter 通常用于合并多个数据源（如多个 SSTable 或内存表）的迭代结果，保证按序输出。
        self.iter.seek_to_first();
        // 若底层迭代器有效（self.iter.valid() 为 true），调用 self.find_next_user_entry(false) 查找第一个符合条件的用户条目（过滤掉内部标记或删除的条目）。
        // 若底层迭代器无效（已无数据），则将当前迭代器标记为无效（self.valid = false）。
        if self.iter.valid() {
            self.find_next_user_entry(
                // skipping=
                false,
            );
        } else {
            self.valid = false;
        }
    }
    fn reset(&mut self) {
        println!("----------db reset----------");
        self.iter.reset();
        self.valid = false;
        self.savedkey.clear();
        self.savedval.clear();
        self.keybuf.clear();
    }
}

/// 这段代码定义了一个 random_period 函数，用于生成一个在特定范围内的随机整数
fn random_period() -> isize {
    // 首先通过 rand::random::<isize>() 生成一个任意大小的 isize 类型随机数（isize 是 Rust 中的有符号整数类型，大小与平台相关）。
    // 然后使用取模运算 % (2 * READ_BYTES_PERIOD) 对随机数进行处理，将其范围限制在 [-2*READ_BYTES_PERIOD + 1, 2*READ_BYTES_PERIOD - 1] 之间（因为有符号整数取模会保留符号）。
    // 结合之前提到的 READ_BYTES_PERIOD 是 1MB（1048576），这个函数的实际效果是生成一个 -2MB 到 2MB 之间的随机整数（不包括 -2MB 和 2MB 本身）。
    rand::random::<isize>() % (2 * READ_BYTES_PERIOD)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::db_impl::testutil::*;
    use crate::db_impl::DB;
    use crate::options;
    use crate::test_util::LdbIteratorIter;
    use crate::types::{current_key_val, Direction};

    use std::collections::HashMap;
    use std::collections::HashSet;
    use std::iter::FromIterator;

    #[test]
    fn db_iter_basic_test() {
        let mut db = build_db().0;
        let mut iter = db.new_iter().unwrap();

        // keys and values come from make_version(); they are each the latest entry.
        let keys: &[&[u8]] = &[
            b"aaa", b"aab", b"aax", b"aba", b"bab", b"bba", b"cab", b"cba",
        ];
        let vals: &[&[u8]] = &[
            b"val1", b"val2", b"val2", b"val3", b"val4", b"val5", b"val2", b"val3",
        ];

        let mut found = vec![];
        for (k, v) in keys.iter().zip(vals.iter()) {
            assert!(iter.advance());
            assert_eq!((k.to_vec(), v.to_vec()), current_key_val(&iter).unwrap());
            let entry = db.get(k).expect("key returned by iterator is in database");
            assert_eq!(v.to_vec(), entry);
            found.push((k.to_vec(), v.to_vec()));
        }

        assert_eq!(found.len(), keys.len());
    }

    #[test]
    fn db_iter_reset() {
        let mut db = build_db().0;
        let mut iter = db.new_iter().unwrap();

        assert!(iter.advance());
        assert!(iter.valid());
        iter.reset();
        assert!(!iter.valid());
        assert!(iter.advance());
        assert!(iter.valid());
    }

    #[test]
    fn db_iter_test_fwd_backwd() {
        let mut db = build_db().0;
        let mut iter = db.new_iter().unwrap();

        // keys and values come from make_version(); they are each the latest entry.
        let keys: &[&[u8]] = &[
            b"aaa", b"aab", b"aax", b"aba", b"bab", b"bba", b"cab", b"cba",
        ];
        let vals: &[&[u8]] = &[
            b"val1", b"val2", b"val2", b"val3", b"val4", b"val5", b"val2", b"val3",
        ];

        // This specifies the direction that the iterator should move to. Based on this, an index
        // into keys/vals is incremented/decremented so that we get a nice test checking iterator
        // move correctness.
        let dirs: &[Direction] = &[
            Direction::Forward,
            Direction::Forward,
            Direction::Forward,
            Direction::Reverse,
            Direction::Reverse,
            Direction::Reverse,
            Direction::Forward,
            Direction::Forward,
            Direction::Reverse,
            Direction::Forward,
            Direction::Forward,
            Direction::Forward,
            Direction::Forward,
        ];
        let mut i = 0;
        iter.advance();
        for d in dirs {
            assert_eq!(
                (keys[i].to_vec(), vals[i].to_vec()),
                current_key_val(&iter).unwrap()
            );
            match *d {
                Direction::Forward => {
                    assert!(iter.advance());
                    i += 1;
                }
                Direction::Reverse => {
                    assert!(iter.prev());
                    i -= 1;
                }
            }
        }
    }

    #[test]
    fn db_iter_test_seek() {
        let mut db = build_db().0;
        let mut iter = db.new_iter().unwrap();

        // gca is the deleted entry.
        let keys: &[&[u8]] = &[b"aab", b"aaa", b"cab", b"eaa", b"aaa", b"iba", b"fba"];
        let vals: &[&[u8]] = &[
            b"val2", b"val1", b"val2", b"val1", b"val1", b"val2", b"val3",
        ];

        for (k, v) in keys.iter().zip(vals.iter()) {
            eprintln!("{:?}", String::from_utf8(k.to_vec()).unwrap());
            iter.seek(k);
            assert_eq!((k.to_vec(), v.to_vec()), current_key_val(&iter).unwrap());
        }

        // seek past last.
        iter.seek(b"xxx");
        assert!(!iter.valid());
        iter.seek(b"aab");
        assert!(iter.valid());

        // Seek skips over deleted entry.
        iter.seek(b"gca");
        assert!(iter.valid());
        assert_eq!(
            (b"gda".to_vec(), b"val5".to_vec()),
            current_key_val(&iter).unwrap()
        );
    }

    #[test]
    fn db_iter_deleted_entry_not_returned() {
        let mut db = build_db().0;
        let mut iter = db.new_iter().unwrap();
        let must_not_appear = b"gca";

        for (k, _) in LdbIteratorIter::wrap(&mut iter) {
            assert!(k.as_slice() != must_not_appear);
        }
    }

    #[test]
    fn db_iter_deleted_entry_not_returned_memtable() {
        let mut db = build_db().0;

        db.put(b"xyz", b"123").unwrap();
        db.delete(b"xyz").unwrap();

        let mut iter = db.new_iter().unwrap();
        let must_not_appear = b"xyz";

        for (k, _) in LdbIteratorIter::wrap(&mut iter) {
            assert!(k.as_slice() != must_not_appear);
        }
    }

    #[test]
    fn db_iter_repeated_open_close() {
        let opt;
        {
            let (mut db, opt_) = build_db();
            opt = opt_;

            db.put(b"xx1", b"111").unwrap();
            db.put(b"xx2", b"112").unwrap();
            db.put(b"xx3", b"113").unwrap();
            db.put(b"xx4", b"114").unwrap();
            db.delete(b"xx2").unwrap();
        }

        {
            let mut db = DB::open("db", opt.clone()).unwrap();
            db.put(b"xx4", b"222").unwrap();
        }

        {
            let mut db = DB::open("db", opt).unwrap();

            let ss = db.get_snapshot();
            // xx5 should not be visible.
            db.put(b"xx5", b"223").unwrap();

            let expected: HashMap<Vec<u8>, Vec<u8>> = HashMap::from_iter(vec![
                (b"xx1".to_vec(), b"111".to_vec()),
                (b"xx4".to_vec(), b"222".to_vec()),
                (b"aaa".to_vec(), b"val1".to_vec()),
                (b"cab".to_vec(), b"val2".to_vec()),
            ]);
            let non_existing: HashSet<Vec<u8>> =
                HashSet::from_iter(vec![b"gca".to_vec(), b"xx2".to_vec(), b"xx5".to_vec()]);

            let mut iter = db.new_iter_at(ss.clone()).unwrap();
            for (k, v) in LdbIteratorIter::wrap(&mut iter) {
                if let Some(ev) = expected.get(&k) {
                    assert_eq!(ev, &v);
                }
                assert!(!non_existing.contains(&k));
            }
        }
    }

    #[test]
    fn db_iter_allow_empty_key() {
        let opt = options::for_test();
        let mut db = DB::open("db", opt).unwrap();
        assert!(db.new_iter().unwrap().next().is_none());
        db.put(&[], &[]).unwrap();
        assert!(db.new_iter().unwrap().next().is_some());
    }
}
